﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Configuration;
using System.Diagnostics;
using System.IO;
using System.Threading;


namespace ProcessFile
{


    public class MultiThreadedProcessor
    {

        public int LineCountToReportUpdate = 25;
        public int ThreadCount = 1;
        public string InputFilename = ""; //ConfigurationManager.AppSettings.Get("inputFile")
        public bool ShowProcessingUpdates = true;

        public void Process(Action<string> processLine )
        {
            var errorLines = new System.Collections.Concurrent.ConcurrentBag<string>();
            
            int processedCount = 0;


            //quickly count lines to know how many
            int totalLineCount = 0;
            while ((ReadNext(null)) != null)
            {
                totalLineCount++;
            }
            RestartAt(0);


            var s = new Stopwatch();
            s.Start();

            int readCount = 0, failedCount = 0;
            string line;
            var queue = new System.Collections.Concurrent.ConcurrentQueue<string>();
            int takenFromQueueCount = 0;

            for (var i = 0; i < ThreadCount; i++)
            {
                var t = new System.Threading.Thread(() =>
                {
                    string lineToProcess = null;

                    try
                    {


                        while (true)
                        {
                            //loop until we get an item, but if not getting anything, let the thread die
                            int failedDequeue = 0;
                            while (!queue.TryDequeue(out lineToProcess))
                            {
                                failedDequeue++;
                                if (failedDequeue > 4)
                                {
                                    return;
                                }
                                System.Threading.Thread.Sleep(1000);
                                
                            }

                            

                            try
                            {

                                processLine(lineToProcess);

                                if (ShowProcessingUpdates)
                                    Console.Write("-"); //put character out when line has been processed

                            }
                            catch (Exception ex)
                            {
                                var msg = ex.Message;
                                errorLines.Add(lineToProcess);
                            }
                            finally
                            {
                                System.Threading.Interlocked.Increment(ref processedCount);
                            }

                        }
                    }
                    catch (Exception ex)
                    {
                        var msg = ex.Message;
                        Console.WriteLine("Catastrophic error on thread  msg={0}", msg);
                        //rethrow exception
                        throw;
                    }
                });

                t.IsBackground = true;
                t.Start();
            }


            while ((line = ReadNext(null)) != null )
            {
                readCount++;

                queue.Enqueue(line);

                takenFromQueueCount = readCount - queue.Count;
                failedCount = errorLines.Count;

                if (readCount % LineCountToReportUpdate == 0 && takenFromQueueCount > 0)
                {
                    double percent = ((double)takenFromQueueCount * 100) / totalLineCount;
                    double msPer = (s.Elapsed.TotalMilliseconds / (takenFromQueueCount));

                    int linesLeft = totalLineCount - takenFromQueueCount;
                    double msLeft = msPer * linesLeft;
                    TimeSpan timeleft = new TimeSpan(0, 0, 0, (int)msLeft / 1000);


                    if (ShowProcessingUpdates)
                    {
                        Console.WriteLine();
                        Console.Write("{0}/{1} {2:##0.###}%", takenFromQueueCount, totalLineCount, percent);
                        Console.WriteLine(" time per: {0:0}ms error={1}", +msPer, failedCount);
                        Console.WriteLine("Total time: {0:hh\\:mm\\:ss}", s.Elapsed);

                        Console.WriteLine("Time remaining (D:HH:MM): {0:0}:{1:00}:{2:00}", timeleft.Days, timeleft.Hours, timeleft.Minutes);
                        Console.WriteLine(" estimated finish time {0:T}", DateTime.Now.Add(timeleft));
                    }

                }
                else
                {
                    if (ShowProcessingUpdates) Console.Write(".");
                }
                Console.Out.FlushAsync();


                //don't fill the queue faster than it is being emptied
                while (queue.Count > LineCountToReportUpdate * 2)
                {
                    Thread.Sleep(1000);
                }

            }

            //wait for queue to empty and all rows have been processed
            while (queue.Count > 0 || processedCount < readCount)
            {
                Thread.Sleep(100);
            }


            s.Stop();
            Console.WriteLine("");
            Console.WriteLine("Total time: {0:hh\\:mm\\:ss}", s.Elapsed);
            Console.WriteLine("Total requests: {0}", readCount);
            Console.WriteLine("Total successes: {0}", readCount - errorLines.Count);
            Console.WriteLine("Total errors: {0}", errorLines.Count);
            Console.WriteLine(" time per: {0:0}ms", (s.Elapsed.TotalMilliseconds / (readCount)));


            if (errorLines.Count > 0)
            {
                Console.WriteLine("Errors received when updating these {0} lines", errorLines.Count);
                foreach (var errorLine in errorLines)
                {
                    Console.WriteLine(" " + errorLine);
                }
            }



        }


        private StreamReader file;

        private string ReadNext(string commentLabel)
        {
            if (file == null)
            {
                file = new StreamReader(InputFilename);
            }

            string line;

            line = file.ReadLine();

            if (line != null && !string.IsNullOrEmpty(commentLabel))
            {
                while (!line.StartsWith("#" + commentLabel + ":") && !line.StartsWith(commentLabel + ":"))
                {
                    if ((line = file.ReadLine()) == null)
                    {
                        throw new Exception("commentLabel: " + commentLabel + " was never found");
                    }
                }

                return line.Substring(commentLabel.Length + 2).Trim();

            }

            while (line != null && (line.StartsWith("#") || string.IsNullOrWhiteSpace(line) ))
            {
                line = file.ReadLine();
            }

            return line;


        }

        private void RestartAt(int startAt)
        {
            file.BaseStream.Seek(startAt, SeekOrigin.Begin);
            file.DiscardBufferedData();

        }


    }


}
